package com.event.ingestion.kafka;

import com.event.ingestion.common.HBaseWriter;
import com.event.ingestion.common.MongoWriter;
import com.event.ingestion.common.Parsable;
import com.event.ingestion.common.Persistable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * @author ZzHh
 * @Classname EventAttendeesConsumer
 * @Description TODO
 * @Date:2020/06/19 11:37
 * @Create by IntelliJ IDEA
 **/

/**
 * EventAttendeesConsumer业务代码编写
 */
public class EventAttendeesConsumer extends BaseConsumer {
//    //hbase parser
//    public static class EventAttendeesHBaseParser implements Parsable<Put> {
//        //event_id,user_id,attend_type
//
//        @Override
//        public Boolean isHeader(String[] fields) {
//            return (isValid(fields) && fields[0].equals("event_id") && fields[1].equals("user_id") && fields[2].equals("attend_type"));
//        }
//
//        @Override
//        public Boolean isValid(String[] fields) {
//            return (fields.length > 2);
//        }
//
//        @Override
//        public Put parse(String[] fields) {
//            Put p = new Put(Bytes.toBytes((fields[0] + "." + fields[1] + "-" + fields[2]).hashCode()));  //rowkey
//
//            p.addColumn(Bytes.toBytes("euat"),Bytes.toBytes("event_id"),Bytes.toBytes(fields[0]));
//            p.addColumn(Bytes.toBytes("euat"),Bytes.toBytes("user_id"),Bytes.toBytes(fields[1]));
//            p.addColumn(Bytes.toBytes("euat"),Bytes.toBytes("attend_type"),Bytes.toBytes(fields[2]));
//
//            return p;
//        }
//    }

    public EventAttendeesConsumer() {
        super(new Persistable [] {new HBaseWriter("events_db:event_attendees",new com.event.ingestion.data.hbase.EventAttendeesParser()),
                                  new MongoWriter("event_attendee",new com.event.ingestion.data.mongo.EventAttendeesParser())
        });
    }

    @Override
    protected int getMaxPollRecords() {
        return 32000;
    }

    @Override
    protected String getKafkaConsumerGrp() {
        return "grpEventAttendees";
    }

    @Override
    protected String getKafkaTopic() {
        return "event_attendees";
    }

    @Override
    protected Boolean getKafkaAutoCommit() {
        return true;
    }
}